#include "utilities.h"

//This is the read thread which will receive the request from other nodes and pass it on to the dispatcher.
void *readfun(void *ptr)
{
	struct timeval readtime;
	int numbytes = -1;
	writeStruct wrStruct = *((writeStruct*) ptr);
	if(printFlag==1)
	{
		printf("Read Thread for Portnumber %d Starts\n",wrStruct.portNumber);
	}
	while(!dieChild)
	{
		allTimers tempTimer;
		char headBuff[27];
		if((numbytes = read(wrStruct.sock, headBuff, 27)) == 0)
		{
			pthread_mutex_lock(&notifyLock);
			notifyCount--;
			pthread_mutex_unlock(&notifyLock);
			//Now clear the map
			pthread_mutex_lock(&writeMapMutex);
			if(writeMap.find(wrStruct.portNumber) != writeMap.end())
			{
				writeMap.erase(writeMap.find(wrStruct.portNumber));
			}
			if(printFlag==1)
			{
				cout<<"eSIZE IN READ "<<writeMap.size()<<endl;
			}
			pthread_mutex_unlock(&writeMapMutex);
			pthread_mutex_lock(&timerlock);
			myTimers.erase(myTimers.find(wrStruct.portNumber));
			pthread_mutex_unlock(&timerlock);
			pthread_mutex_lock(&maplock);
			myMap.erase(myMap.find(wrStruct.portNumber));
			pthread_mutex_unlock(&maplock);
			//Now clean shutdown the system
			pthread_mutex_lock(&notifyLock);
			if(notifyCount == 0)
			{
				serverShutdown = true;
				dieChild = true;
				if(printFlag==1)
				{
					cout<<statusFile<<endl;
				}
				close(globalAcceptSock);
			}
			pthread_mutex_unlock(&notifyLock);
			if(printFlag==1)
			{
				cout<<"Error while reading"<<endl;
			}
			close(wrStruct.sock);
			pthread_exit(NULL);
		}
		if(numbytes == -1)
		{
			//Now clear the map
			pthread_mutex_lock(&writeMapMutex);
			if(writeMap.find(wrStruct.portNumber) != writeMap.end())
			{
				writeMap.erase(writeMap.find(wrStruct.portNumber));
			}
			if(printFlag==1)
			{
				cout<<"SIZE IN READ "<<writeMap.size()<<endl;
			}
			pthread_mutex_unlock(&writeMapMutex);
			pthread_mutex_lock(&timerlock);
			myTimers.erase(myTimers.find(wrStruct.portNumber));
			pthread_mutex_unlock(&timerlock);
			pthread_mutex_lock(&maplock);
			myMap.erase(myMap.find(wrStruct.portNumber));
			pthread_mutex_unlock(&maplock);
			if(printFlag==1)
			{
				cout<<"Owned socket is closed - "<<wrStruct.sock<<endl;
			}
			close(wrStruct.sock);
			pthread_exit(NULL);
		}
		if(dieChild)
		{
			close(wrStruct.sock);
			pthread_exit(NULL);
		}
		pthread_mutex_lock(&timerlock);
		tempTimer = myTimers[wrStruct.portNumber];
		tempTimer.keepAliveTime = 0;
		myTimers[wrStruct.portNumber] = tempTimer;
		pthread_mutex_unlock(&timerlock);
		msg_header temp;
		fillMsgHdrStructure(temp, headBuff);
		
		if(temp.msgtype == 0xCC || temp.msgtype == 0xDB)
		{
			if(printFlag == 1)
			{
				printf("READ TYPE - %02x\n",(uint8_t)headBuff[0]);
			}
			int f = 0;
			int logbuffflag = 0;
			char logdataBuff[8192];
			memset(&logdataBuff, '\0', 8192);
			int forCounter = temp.datalength/(8192);
			int extraCounter = temp.datalength%(8192);
			if(forCounter == 0)
			{
				if(!notifyFlag)
				{
					pthread_mutex_lock(&eventQueueMutex);
					listBuffStruct lbtemp;
					memset(&lbtemp.listBuff, '\0', 8192);
					memcpy(&lbtemp.listBuff, headBuff, sizeof(headBuff));
					lbtemp.datalength = 27;
					lbtemp.portNumber = wrStruct.portNumber;
					eventQueue.push_back(lbtemp);
					char dataBuff[temp.datalength];
					memset(dataBuff,'\0',temp.datalength);
					if((numbytes = read(wrStruct.sock, dataBuff, temp.datalength)) == 0)
					{
						pthread_mutex_lock(&notifyLock);
						notifyCount--;
						pthread_mutex_unlock(&notifyLock);
						//Now clear the map
						pthread_mutex_lock(&writeMapMutex);
						if(writeMap.find(wrStruct.portNumber) != writeMap.end())
						{
							writeMap.erase(writeMap.find(wrStruct.portNumber));
						}
						pthread_mutex_unlock(&writeMapMutex);
						pthread_mutex_lock(&timerlock);
						myTimers.erase(myTimers.find(wrStruct.portNumber));
						pthread_mutex_unlock(&timerlock);
						pthread_mutex_lock(&maplock);
						myMap.erase(myMap.find(wrStruct.portNumber));
						pthread_mutex_unlock(&maplock);
						//Now clean shutdown the system
						pthread_mutex_lock(&notifyLock);
						if(notifyCount == 0)
						{
							
							serverShutdown = true;
							dieChild = true;
							if(printFlag==1)
							{
								cout<<statusFile<<endl;
							}
							close(globalAcceptSock);
						}
						pthread_mutex_unlock(&notifyLock);
						if(printFlag==1)
						{
							cout<<"Error while reading"<<endl;
						}
						close(wrStruct.sock);
						pthread_exit(NULL);
					}
					if(numbytes == -1)
					{
						//Now clear the map
						pthread_mutex_lock(&writeMapMutex);
						if(writeMap.find(wrStruct.portNumber) != writeMap.end())
						{
							writeMap.erase(writeMap.find(wrStruct.portNumber));
						}
						pthread_mutex_unlock(&writeMapMutex);
						pthread_mutex_lock(&timerlock);
						myTimers.erase(myTimers.find(wrStruct.portNumber));
						pthread_mutex_unlock(&timerlock);
						pthread_mutex_lock(&maplock);
						myMap.erase(myMap.find(wrStruct.portNumber));
						pthread_mutex_unlock(&maplock);
						if(printFlag==1)
						{
							cout<<"Owned socket is closed"<<endl;
						}
						close(wrStruct.sock);
						pthread_exit(NULL);
					}
					if(dieChild)
					{
						close(wrStruct.sock);
						pthread_exit(NULL);
					}
						
					memset(&lbtemp.listBuff, '\0', 8192);
					memcpy(lbtemp.listBuff, dataBuff, temp.datalength);
					if(logbuffflag==0)
					{
						memcpy(logdataBuff,dataBuff,30);
						logbuffflag=1;
					}
					lbtemp.datalength = temp.datalength;
					lbtemp.portNumber = wrStruct.portNumber;
					eventQueue.push_back(lbtemp);
					pthread_mutex_unlock(&eventQueueMutex);
				}
			}
			else
			{
				if(!notifyFlag)
				{
					pthread_mutex_lock(&eventQueueMutex);
					listBuffStruct lbtemp;
					memset(&lbtemp.listBuff, '\0', 8192);
					memcpy(&lbtemp.listBuff, headBuff, sizeof(headBuff));
					lbtemp.datalength = 27;
					lbtemp.portNumber = wrStruct.portNumber;
					eventQueue.push_back(lbtemp);
					char dataBuff[8192];
					for(f=0;f<forCounter;f++)
					{
						memset(dataBuff,'\0',8192);
						if((numbytes = read(wrStruct.sock, dataBuff, 8192)) == 0)
						{
							pthread_mutex_lock(&notifyLock);
							notifyCount--;
							pthread_mutex_unlock(&notifyLock);
							//Now clear the map
							pthread_mutex_lock(&writeMapMutex);
							if(writeMap.find(wrStruct.portNumber) != writeMap.end())
							{
								writeMap.erase(writeMap.find(wrStruct.portNumber));
							}
							pthread_mutex_unlock(&writeMapMutex);
							pthread_mutex_lock(&timerlock);
							myTimers.erase(myTimers.find(wrStruct.portNumber));
							pthread_mutex_unlock(&timerlock);
							pthread_mutex_lock(&maplock);
							myMap.erase(myMap.find(wrStruct.portNumber));
							pthread_mutex_unlock(&maplock);
							//Now clean shutdown the system
							pthread_mutex_lock(&notifyLock);
							if(notifyCount == 0)
							{
								
								serverShutdown = true;
								dieChild = true;
								if(printFlag==1)
								{
									cout<<statusFile<<endl;
								}
								close(globalAcceptSock);
							}
							pthread_mutex_unlock(&notifyLock);
							if(printFlag==1)
							{
								cout<<"Error while reading"<<endl;
							}
							close(wrStruct.sock);
							pthread_exit(NULL);
						}
						if(numbytes == -1)
						{
							//Now clear the map
							pthread_mutex_lock(&writeMapMutex);
							if(writeMap.find(wrStruct.portNumber) != writeMap.end())
							{
								writeMap.erase(writeMap.find(wrStruct.portNumber));
							}
							pthread_mutex_unlock(&writeMapMutex);
							pthread_mutex_lock(&timerlock);
							myTimers.erase(myTimers.find(wrStruct.portNumber));
							pthread_mutex_unlock(&timerlock);
							pthread_mutex_lock(&maplock);
							myMap.erase(myMap.find(wrStruct.portNumber));
							pthread_mutex_unlock(&maplock);
							if(printFlag==1)
							{
								cout<<"Owned socket is closed"<<endl;
							}
							close(wrStruct.sock);
							pthread_exit(NULL);
						}
						if(dieChild)
						{
							close(wrStruct.sock);
							pthread_exit(NULL);
						}
						memset(&lbtemp.listBuff, '\0', 8192);
						memcpy(lbtemp.listBuff, dataBuff,8192);
						if(printFlag == 1)
						{
							printf("%s\n",&(dataBuff[24]));
						}
						if(logbuffflag==0)
						{
							memcpy(logdataBuff,dataBuff,30);
							logbuffflag=1;
						}
						lbtemp.datalength = 8192;
						lbtemp.portNumber = wrStruct.portNumber;
						eventQueue.push_back(lbtemp);
					}
					if(extraCounter!=0)
					{
						
						memset(dataBuff,'\0',8192);
						if((numbytes = read(wrStruct.sock, dataBuff, extraCounter)) == 0)
						{
							pthread_mutex_lock(&notifyLock);
							notifyCount--;
							pthread_mutex_unlock(&notifyLock);
							//Now clear the map
							pthread_mutex_lock(&writeMapMutex);
							if(writeMap.find(wrStruct.portNumber) != writeMap.end())
							{
								writeMap.erase(writeMap.find(wrStruct.portNumber));
							}
							pthread_mutex_unlock(&writeMapMutex);
							pthread_mutex_lock(&timerlock);
							myTimers.erase(myTimers.find(wrStruct.portNumber));
							pthread_mutex_unlock(&timerlock);
							pthread_mutex_lock(&maplock);
							myMap.erase(myMap.find(wrStruct.portNumber));
							pthread_mutex_unlock(&maplock);
							//Now clean shutdown the system
							pthread_mutex_lock(&notifyLock);
							if(notifyCount == 0)
							{
								serverShutdown = true;
								dieChild = true;
								if(printFlag==1)
								{
									cout<<statusFile<<endl;
								}
								close(globalAcceptSock);
							}
							pthread_mutex_unlock(&notifyLock);
							if(printFlag==1)
							{
								cout<<"Error while reading"<<endl;
							}
							close(wrStruct.sock);
							pthread_exit(NULL);
						}
						if(numbytes == -1)
						{
							//Now clear the map
							pthread_mutex_lock(&writeMapMutex);
							if(writeMap.find(wrStruct.portNumber) != writeMap.end())
							{
								writeMap.erase(writeMap.find(wrStruct.portNumber));
							}
							pthread_mutex_unlock(&writeMapMutex);
							pthread_mutex_lock(&timerlock);
							myTimers.erase(myTimers.find(wrStruct.portNumber));
							pthread_mutex_unlock(&timerlock);
							pthread_mutex_lock(&maplock);
							myMap.erase(myMap.find(wrStruct.portNumber));
							pthread_mutex_unlock(&maplock);
							if(printFlag==1)
							{
								cout<<"Owned socket is closed"<<endl;
							}
							close(wrStruct.sock);
							pthread_exit(NULL);
						}
						if(dieChild)
						{
							close(wrStruct.sock);
							pthread_exit(NULL);
						}
						memset(&lbtemp.listBuff, '\0', extraCounter);
						memcpy(lbtemp.listBuff, dataBuff,extraCounter);
						lbtemp.datalength = extraCounter;
						lbtemp.portNumber = wrStruct.portNumber;
						eventQueue.push_back(lbtemp);
					}
					pthread_mutex_unlock(&eventQueueMutex);
				}
			}
			gettimeofday(&readtime,NULL);
			listBuffStruct lbtemp;
			memset(&lbtemp.listBuff, '\0', 8192);
			memcpy(&lbtemp.listBuff, headBuff, sizeof(headBuff));
			memcpy(&(lbtemp.listBuff[27]), logdataBuff, 30);
			lbtemp.datalength = 27+temp.datalength;
			lbtemp.portNumber = wrStruct.portNumber;
			write_to_log(lbtemp,0,readtime);
		}
		else
		{
			char dataBuff[temp.datalength];
			if(temp.datalength != 0)
			{
				if((numbytes = read(wrStruct.sock, dataBuff, temp.datalength)) == 0)
				{
					pthread_mutex_lock(&notifyLock);
					notifyCount--;
					pthread_mutex_unlock(&notifyLock);
					//Now clear the map
					pthread_mutex_lock(&writeMapMutex);
					if(writeMap.find(wrStruct.portNumber) != writeMap.end())
					{
						writeMap.erase(writeMap.find(wrStruct.portNumber));
					}
					pthread_mutex_unlock(&writeMapMutex);
					pthread_mutex_lock(&timerlock);
					myTimers.erase(myTimers.find(wrStruct.portNumber));
					pthread_mutex_unlock(&timerlock);
					pthread_mutex_lock(&maplock);
					myMap.erase(myMap.find(wrStruct.portNumber));
					pthread_mutex_unlock(&maplock);
					//Now clean shutdown the system
					pthread_mutex_lock(&notifyLock);
					if(notifyCount == 0)
					{
						
						serverShutdown = true;
						dieChild = true;
						if(printFlag==1)
						{
							cout<<statusFile<<endl;
						}
						close(globalAcceptSock);
					}
					pthread_mutex_unlock(&notifyLock);
					if(printFlag==1)
					{
						cout<<"Error while reading"<<endl;
					}
					close(wrStruct.sock);
					pthread_exit(NULL);
				}
				if(numbytes == -1)
				{
					//Now clear the map
					pthread_mutex_lock(&writeMapMutex);
					if(writeMap.find(wrStruct.portNumber) != writeMap.end())
					{
						writeMap.erase(writeMap.find(wrStruct.portNumber));
					}
					pthread_mutex_unlock(&writeMapMutex);
					pthread_mutex_lock(&timerlock);
					myTimers.erase(myTimers.find(wrStruct.portNumber));
					pthread_mutex_unlock(&timerlock);
					pthread_mutex_lock(&maplock);
					myMap.erase(myMap.find(wrStruct.portNumber));
					pthread_mutex_unlock(&maplock);
					if(printFlag==1)
					{
						cout<<"Owned socket is closed"<<endl;
					}
					close(wrStruct.sock);
					pthread_exit(NULL);
				}
				if(dieChild)
				{
					close(wrStruct.sock);
					pthread_exit(NULL);
				}
			}
			//Do not put in the dispatcher queue if the notify is sent for shutdown
			if(!notifyFlag)
			{
				listBuffStruct lbtemp;
				gettimeofday(&readtime,NULL);
				memset(&lbtemp.listBuff, '\0', 8192);
				memcpy(&lbtemp.listBuff, headBuff, sizeof(headBuff));
				memcpy(&(lbtemp.listBuff[27]), dataBuff, temp.datalength);
				lbtemp.datalength = 27+temp.datalength;
				lbtemp.portNumber = wrStruct.portNumber;
				write_to_log(lbtemp,0,readtime);
				pthread_mutex_lock(&eventQueueMutex);
				eventQueue.push_back(lbtemp);
				pthread_mutex_unlock(&eventQueueMutex);	
			}
		}
	}
	close(wrStruct.sock);
	pthread_exit(NULL);
}
